package com.intct.flink.study;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-14 14:08
 */
public class SQLTest10 {
    public static void main(String[] args) {
        // 配置属性
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8081");

        // 1 获取取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 并行度
        env.setParallelism(1);
        // 开启检点
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 获取SQL执行环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 映射mysql的订单
        tenv.executeSql(
                "create table orders( " +
                        // 如果是Flink sql的关键字，需要用反引号转义
                        "  `user`        BIGINT " +
                        "  ,product        STRING " +
                        "  ,order_time    TIMESTAMP(3) " +
                        "  ,customer_id    INT " +
                        "  ,order_id    INT " +
                        // 使用proctime获取数据处理时间
                        "  ,proc_time as proctime() " +
                        // 水位线
                        "  ,watermark for order_time as order_time - interval '0' second " +
                        // 设置主键
                        "    ,PRIMARY KEY (`user`) NOT ENFORCED " +
                        ") with ( " +
                        "  'connector' = 'mysql-cdc' " +
                        "  ,'hostname' = 'cdh-node' " +
                        "  ,'port' = '13306' " +
                        "  ,'username' = 'root' " +
                        "  ,'password' = 'Test_090110' " +
                        "  ,'database-name' = 'm1' " +
                        "  ,'table-name' = 'orders' " +
                        ") ");

        // 映射hbase的客户维度表
        tenv.executeSql(
                "create table dim_customers(" +
                        " id STRING" +
                        " ,f1 ROW<name STRING" +
                        "     ,country STRING" +
                        "        ,zip STRING>" +
                        " ,PRIMARY KEY (id) NOT ENFORCED" +
                        ") with (" +
                        " 'connector' = 'hbase-2.2'" +
                        // 配置HBase表时，需要注意增加命名空间名称。如果在default命名空间名称，不需要命名空间名称
                        "    ,'table-name' = 'intct85:dim_customers'" +
                        "    ,'zookeeper.quorum' = 'cdh-node:2181'" +
                        // lookup关联维度表，可以开启异步读取  后续还有可以缓存，具体看官网帮助
                        "    ,'lookup.async' = 'true'" +
                        ")");
        // 退维==打宽
        tenv.executeSql(
                "select" +
                        "   o.`user`" +
                        "   ,o.product" +
                        "   ,o.order_time" +
                        "   ,o.customer_id" +
                        "   ,o.order_id" +
                        // 显示HBase列名时，使用格式：表名称.列族名称.列限定符 == 列名
                        "   ,d.f1.name" +
                        "   ,d.f1.country" +
                        "   ,d.f1.zip" +
                        " from orders as o " +
                        // 使用lookup方式关联维度时，固定：维度名称 for system_time as of 必须处理时间 as 表别名
                        " join dim_customers for system_time as of o.proc_time as d " +
                        // 类型转换：将数字转换为 字符串
                        "   on cast(o.customer_id as string) = d.id").print();

    }
}
